package com.thiswhyme.republish.rabbitmq.core;

import com.thiswhyme.republish.common.UidGenService;
import com.thiswhyme.republish.rabbitmq.config.BindingPlusProperties;
import com.thiswhyme.republish.rabbitmq.config.BindingServicePlusProperties;
import com.thiswhyme.republish.rabbitmq.core.domain.MessageInfo;
import com.thiswhyme.republish.rabbitmq.core.domain.MessageType;
import com.thiswhyme.republish.rabbitmq.core.domain.PublishMqHeaders;
import com.thiswhyme.republish.rabbitmq.core.domain.PublishStatus;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.Optional;

/**
 * @description: 可重试的消息监听抽象类
 * 注：消息内容类型：String，contentType：application/json，deliveryMode：持久化
 * @author: thiswhyme
 * @date: 2021/7/27
 */
@Slf4j
public abstract class AbstractRetriableReceiver implements IRetriableReceiver {

    @Autowired
    private BindingServicePlusProperties bindingServicePlusProperties;

    @Autowired
    private MessageInfoService messageInfoService;

    @Autowired
    private UidGenService uidGenService;

    @Autowired
    private RepublishRedisOperation republishRedisOperation;

    /**
     * 消息重发
     * @param message 收到的消息对象（必填）
     * @param channelName 消息通道名
     * @param bizParams 业务参数，用于重试n次后仍然失败的自定义业务处理（非必填）
     * @throws Exception
     */
    public void retry(Message<String> message, String channelName, Object bizParams) {
        //参数校验
        String exchangeName = (String) message.getHeaders().get(AmqpHeaders.RECEIVED_EXCHANGE);
        if (!StringUtils.hasText(exchangeName)) {
            log.warn("【消费重试处理器】-未获取到交换机名，跳过处理，消息UUID：{}", message.getHeaders().getId());
            return;
        }
        String routingKey = (String) message.getHeaders().get(AmqpHeaders.RECEIVED_ROUTING_KEY);
        if (!StringUtils.hasText(routingKey)) {
            log.warn("【消费重试处理器】-未获取到交路由键，跳过处理，消息UUID：{}", message.getHeaders().getId());
            return;
        }
        if(CollectionUtils.isEmpty(bindingServicePlusProperties.getBindings())) {
            log.warn("【消费重试处理器】-未获取到重试配置，跳过处理，消息UUID：{}", message.getHeaders().getId());
            return;
        }
        BindingPlusProperties properties = bindingServicePlusProperties.getBindings().get(channelName);
        if(null == properties) {
            log.warn("【消费重试处理器】-未获取到重试配置，跳过处理，消息UUID：{}", message.getHeaders().getId());
            return;
        }

        doRetry(channelName, message.getHeaders(), message.getPayload(), properties, bizParams);
    }

    /**
     * 消息重发
     * @param headers 消息头
     * @param msgContent 待重发的消息内容
     * @param properties 配置
     * @param bizParams 自定义消息头
     * @throws IOException
     */
    private void doRetry(String channelName, MessageHeaders headers, String msgContent, BindingPlusProperties properties,
                         Object bizParams) {
        Long uniqueId = (Long) headers.get(PublishMqHeaders.HEADER_RETRY_UNIQUE_ID);
        int currentRetryTimes = (int) Optional.ofNullable(headers.get(PublishMqHeaders.HEADER_RECEIVE_RETRY_TIMES)).orElse(0);

        if (currentRetryTimes < properties.getRetryTimes()){
            //未超过重试次数，获取消息（不存在则创建），将消息ID置入待重发zset
            MessageInfo messageInfo = null;
            if (null != uniqueId) {
                messageInfo = messageInfoService.queryById(uniqueId);
            }
            if (null == messageInfo) {
                uniqueId = uidGenService.fetchUid();
                messageInfoService.recordMessage(uniqueId, channelName, String.class.getName(), headers, msgContent,
                        PublishStatus.CONSUME_TO_ACK, MessageType.CONSUME);
            }
            republishRedisOperation.addToRepublishZSet(uniqueId, System.currentTimeMillis()
                    + Optional.ofNullable(properties.getDelayRetryMillis().get(currentRetryTimes)).orElse(1000));
        } else {
            //已超过重试次数，更新状态并执行自定义逻辑
            log.info("重试{}次后仍然失败，开始自定义处理逻辑，消息内容：{}", properties.getRetryTimes(), msgContent);
            messageInfoService.updatePublishStatus(uniqueId, PublishStatus.FAIL);
            doIfRetryFail(bizParams);
        }
    }
}
